package com.dongdong.mq;



import com.dongdong.api.SMSApi;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.UUID;

@Component
@RocketMQMessageListener(consumerGroup = "xiaohuihui", topic = "SMS")
public class Consumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private SMSApi smsApi;
    @Override
    public void onMessage(String s) {
        System.out.println("消费消息:" + s);

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = list.get(0);
                String msg = new String(message.getBody());

                smsApi.getCode(msg);

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;


/*//                重试次数
                int reconsumeTimes = message.getReconsumeTimes();
                try {
//                    int i = 1 / 0;
//                    消息重复问题（消息的幂等性）
                    Boolean queue = redisTemplate.hasKey("queue");
                    if (!queue) {
                        System.out.println("我的小朋友:" + msg);
                        redisTemplate.boundValueOps("queue").set(UUID.randomUUID().toString());
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {

//                    ack确认机制
                    System.out.println("当前重试次数为：--------->>"+reconsumeTimes);
                    if (reconsumeTimes > 1) {
                        System.out.println("重试次数大于2次，直接进入死信队列表.");
//                        通知RocketMQ不需要重试了
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }*/
            }
        });
    }
}
